package com.yujiahao.bigdata.streaming


import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import java.util.Date
import java.io.{File, PrintWriter}
import java.text.SimpleDateFormat
import scala.collection.mutable.ListBuffer

//TODO 需求三：最近一分钟 某个广告点击量趋势统计
object Stream_Source_Test_Kafka3 {
  def main(args: Array[String]): Unit = {

    //TODO SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))


    //TODO 4、业务逻辑-- 从kafka中采集数据
    //在流式数据处理中，kafka应用特别多，所以很多的框架都有相应的工具类

    //4.1.定义Kafka参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu", //消费者组名
      //KV反序列化
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //4.2、kafka消费参数
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String]( //这个是读取Kafka的KV类型
        ssc, //上面的环境对象
        LocationStrategies.PreferConsistent, //位置策略，由框架自行选择
        //消费策略
        ConsumerStrategies.Subscribe[String, String](Set("atguigu210819"), kafkaPara))

    //因为kafka中K和分区策略有关系，因此这里数据直接就Values
    val kafkaValuse: DStream[String] = kafkaDStream.map(_.value())
    //处理kafka的数据
    kafkaValuse.map(
      data => {
        //这里应为时间是毫秒，要想换算成秒，并且是整数点，因此就要做一点处理
        val datas: Array[String] = data.split(" ")
        (datas(0).toLong / 10000 * 10000, 1)
      }
    ).window(
      Seconds(60),
      Seconds(10)
    ).reduceByKey(_ + _)
      //将结果生成json 文件
      .foreachRDD(

        rdd => {
          //先拿到数据
          val datas: Array[(Long, Int)] = rdd.collect()
          //在进行排序，因为最终是按照时间进行统计的升序
          val sourDatas: Array[(Long, Int)] = datas.sortBy(_._1)
          //声明一个集合将数据追加在里面
          val list: ListBuffer[String] = ListBuffer[String]()

          sourDatas.foreach{
            case (time , cnt) =>{
              //因为这里的时间是毫秒所以要将时间转换一下格式
              val times: SimpleDateFormat = new SimpleDateFormat("mm:ss")

             //这里将数据结果进行追加以及按照格式进行拼接
              list.append(s""" { "xtime":"${times.format(new Date(time))}", "yval":"${cnt}" }""")
            }
          }

          //接着将数据写到文件中，Scala中没有写文件操作，Java中有,所以给一个路径将数据保存到文件里面去
          val out: PrintWriter = new PrintWriter(new File("D:\\Atguigu\\05_Code\\14_Spark\\02_MySpark\\data\\adclick\\adclick.json"))
        //这里将数据进行打印，按照JSON文件的格式拼接一下，这里将数据专门放在一个list集合处理了
          out.print(s"[${list.mkString(",")}]")
          out.flush()
          out.close()
        }
      )

    //TODO 2、启动采集器
    ssc.start()
    //TODO 3、Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }

}
